// Copyright 2022 Matrix Origin
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package ctl

import (
	"fmt"
	"strconv"
	"strings"
	"unicode/utf8"

	"github.com/matrixorigin/matrixone/pkg/container/types"

	"github.com/matrixorigin/matrixone/pkg/clusterservice"
	"github.com/matrixorigin/matrixone/pkg/common/moerr"
	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
	"github.com/matrixorigin/matrixone/pkg/pb/query"
	"github.com/matrixorigin/matrixone/pkg/vm/engine"
	"github.com/matrixorigin/matrixone/pkg/vm/process"
)

// force build remote datasource
// mo_ctl("cn", "reader", "enable:force_build_remote_ds:tid1,tid2");
// mo_ctl("cn", "reader", "disable:force_build_remote_ds:tid1,tid2");
//
// force shuffle specified number blocks and table
// mo_ctl("cn", "reader", "enable:force_shuffle:tid1,tid2:blk_cnt");
// mo_ctl("cn", "reader", "disable:force_shuffle:tid1,tid2:blk_cnt");

func handleCtlReader(
	proc *process.Process,
	service serviceType,
	parameter string,
	sender requestSender,
) (Result, error) {
	if service != cn {
		return Result{}, moerr.NewWrongServiceNoCtx("only cn supported", string(service))
	}

	parameter = strings.ToLower(parameter)
	args := strings.Split(parameter, ":")

	cns := make([]string, 0)
	clusterservice.GetMOCluster(proc.GetService()).GetCNService(clusterservice.Selector{}, func(cn metadata.CNService) bool {
		cns = append(cns, cn.ServiceID)
		return true
	})

	info := map[string]string{}
	for idx := range cns {
		// the current cn also need to process this span cmd
		if cns[idx] == proc.GetQueryClient().ServiceID() {
			info[cns[idx]] = UpdateCurrentCNReader(args[0], args[1:]...)
		} else {
			// transfer query to another cn and receive its response
			request := proc.GetQueryClient().NewRequest(query.CmdMethod_CtlReader)
			extra := args[2]
			if len(args) == 4 {
				extra += fmt.Sprintf(":%s", args[3])
			}

			request.CtlReaderRequest = &query.CtlReaderRequest{
				Cmd:   args[0],
				Cfg:   args[1],
				Extra: types.EncodeStringSlice([]string{extra}),
			}
			resp, err := TransferRequest2OtherCNs(proc, cns[idx], request)
			if resp == nil || err != nil {
				// no such cn service
				info[cns[idx]] = fmt.Sprintf("transfer to %s failed, err: %v", cns[idx], err)
			} else if resp.CtlReaderResponse != nil {
				info[cns[idx]] = resp.CtlReaderResponse.Resp
			}
		}
	}

	data := ""
	for k, v := range info {
		data += fmt.Sprintf("%s:%s; ", k, v)
	}

	return Result{
		Method: CtlReaderMethod,
		Data:   data,
	}, nil
}

func UpdateCurrentCNReader(cmd string, cfgs ...string) string {
	cc := false
	if cmd == "enable" {
		cc = true
	} else if cmd == "disable" {
		cc = false
	} else {
		return fmt.Sprintf("not support cmd: %s", cmd)
	}

	if len(cfgs) == 2 && cfgs[0] == "force_build_remote_ds" {
		tbls := strings.Split(cfgs[1], ",")
		engine.SetForceBuildRemoteDS(cc, tbls)

	} else if len(cfgs) == 3 && cfgs[0] == "force_shuffle" {
		tbls := strings.Split(cfgs[1], ",")
		blkCnt, err := strconv.Atoi(cfgs[2])
		if err != nil {
			return fmt.Sprintf("parametrer err: %v", err)
		}

		engine.SetForceShuffleReader(cc, tbls, blkCnt)
	} else {
		return fmt.Sprintf("not support cfg: %v", cfgs)
	}

	return fmt.Sprintf("successed cmd: %s, cfg: %v", cmd, cfgs)
}

// handlePrefetchOnSubscribed distributes user-provided regex patterns to all CNs
// and updates disttae.Engine's PrefetchOnSubscribed configuration on each CN.
// `parameter` accepts comma or newline separated patterns, each optionally
// wrapped in single or double quotes, e.g.:
//
//	'^mo_catalog\.mo_tables$', '^mysql\..*$', '^test1\..*$', '^test2\.t1$'
//
// 'Clean All' to clean all patterns
func handlePrefetchOnSubscribed(
	proc *process.Process,
	service serviceType,
	parameter string,
	sender requestSender,
) (Result, error) {
	if service != cn {
		return Result{}, moerr.NewWrongServiceNoCtx("only cn supported", string(service))
	}

	patterns, err := parsePrefetchOnSubscribed(parameter)
	if err != nil {
		return Result{}, err
	}

	cns := make([]string, 0)
	clusterservice.GetMOCluster(proc.GetService()).GetCNService(clusterservice.Selector{}, func(cn metadata.CNService) bool {
		cns = append(cns, cn.ServiceID)
		return true
	})

	info := map[string]string{}
	for idx := range cns {
		if cns[idx] == proc.GetQueryClient().ServiceID() {
			info[cns[idx]] = UpdateCurrentCNPrefetchOnSubscribed(patterns)
			continue
		}

		request := proc.GetQueryClient().NewRequest(query.CmdMethod_CtlPrefetchOnSubscribed)
		request.CtlPrefetchOnSubscribedRequest = &query.CtlPrefetchOnSubscribedRequest{
			Patterns: patterns,
		}

		resp, err := TransferRequest2OtherCNs(proc, cns[idx], request)
		if resp == nil || err != nil {
			info[cns[idx]] = fmt.Sprintf("transfer to %s failed, err: %v", cns[idx], err)
			continue
		}
		if resp.CtlPrefetchOnSubscribedResponse != nil {
			info[cns[idx]] = resp.CtlPrefetchOnSubscribedResponse.Resp
		}
	}

	data := ""
	for k, v := range info {
		data += fmt.Sprintf("%s:%s; ", k, v)
	}

	return Result{
		Method: PrefetchOnSubscribed,
		Data:   data,
	}, nil
}

func UpdateCurrentCNPrefetchOnSubscribed(patterns []string) string {
	if err := engine.SetPrefetchOnSubscribed(patterns); err != nil {
		return fmt.Sprintf("set prefetch_on_subscribed failed: %v", err)
	}
	return fmt.Sprintf("prefetch_on_subscribed updated, patterns: %d", len(patterns))
}

func parsePrefetchOnSubscribed(parameter string) ([]string, error) {
	if len(strings.TrimSpace(parameter)) == 0 {
		return nil, moerr.NewInvalidInputNoCtx("prefetch_on_subscribed requires at least one pattern")
	}

	var result []string
	var start int
	var quote rune
	input := parameter

	appendToken := func(token string) {
		token = strings.TrimSpace(token)
		if token == "" {
			return
		}
		runes := []rune(token)
		if len(runes) >= 2 && runes[0] == runes[len(runes)-1] && (runes[0] == '\'' || runes[0] == '"') {
			token = string(runes[1 : len(runes)-1])
		}
		if token != "" {
			result = append(result, token)
		}
	}

	for i, r := range input {
		switch r {
		case '\'', '"':
			if quote == 0 {
				quote = r
			} else if quote == r {
				quote = 0
			}
		case ',', '\n', '\r':
			if quote == 0 {
				appendToken(input[start:i])
				start = i + 1
			}
		}
	}
	if start <= len(input) {
		appendToken(input[start:])
	}

	if len(result) == 0 {
		return nil, moerr.NewInvalidInputNoCtx("prefetch_on_subscribed requires at least one pattern")
	}
	for _, pattern := range result {
		if !utf8.ValidString(pattern) {
			return nil, moerr.NewInvalidInputNoCtxf("invalid utf-8 pattern: %s", pattern)
		}
	}
	return result, nil
}
